package net.bwie.realtime.jtp.dws;



import jdk.javadoc.internal.doclets.formats.html.Table;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

import java.util.List;

/**
 *搜索关键词实时统计，其中使用IkAnalyser分词器进行分词，采用FlinkSQL方式数据处理。
 * @author xuanyu
 * @date 2025/5/20
 */
public class JtpTrafficSearchKeywordMinuteWindowDwsJob {

    public static void main(String[] args) {
        // 1. 表执行环境
        TableEnvironment tabEnv = getTableEnv() ;

        // 2. 输入表-input：映射到Kafka消息队列
        createInputTable(tabEnv);

        // 3. 数据处理-select
        Table reportTable = handle(tabEnv);

        // 4. 输出表-output：映射到Doris表
        createOutputTable(tabEnv) ;

        // 5. 保存数据-insert
        saveToSink(tabEnv, reportTable) ;
    }

    private static void saveToSink(TableEnvironment tabEnv, Table reportTable) {
        // a. 注册Table为表，给以表名称

        // b. 查询-插入

    }

    private static void createOutputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql(
                ""
        );
    }

    private static Table handle(TableEnvironment tabEnv) {
        // s1-获取搜索词和搜索时间

        // s2-使用自定义UDTF函数，对搜索词进行中文分词

        // s3-设置窗口进行分组、聚合计算

        // 返回计算结果

        return null;
    }

    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("");
    }


    private static TableEnvironment getTableEnv(){
        // 1环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings) ;
        // 2配置属性设置
        Confignet.bwie.realtime.jtp.dwsuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        // 3返回对象
        return tabEnv;
    }


    /**
     * FlinkSQL中自定义UDTF函数，使用IKAnalyzer分词器对搜索词进行分词
     https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/functions/udfs/#table-functions
     */
    @FunctionHint(output = @DataTypeHint("ROW<keyword STRING>"))
    public static class IkAnalyzerFunction extends TableFunction<Row> {
        public void eval(String fullWord) throws Exception {
            // 中文分词
            List<String> list = AnalyzerUtil.ikAnalyzer(fullWord);
            // 循环遍历输出
            for (String keyword : list) {
                collect(Row.of(keyword));
            }
        }
    }


}
